package cn.iotab.iotkit.transport.api.driver;

import cn.iotab.iotkit.transport.api.common.component.SingleThreadScheduler;
import cn.iotab.iotkit.transport.api.common.queue.KafkaTransportQueueFactory;
import lombok.Getter;

import java.util.concurrent.atomic.AtomicBoolean;

public class TransportDriver {

    private final AtomicBoolean started = new AtomicBoolean(false);

    private final TransportDriverConfig transportDriverConfig;

    //component
    @Getter
    private SingleThreadScheduler singleThreadScheduler;
    //component
    @Getter
    private KafkaTransportQueueFactory queueFactory;
    //component
    @Getter
    private DefaultTransportDriverService transportDriverService;

    /**
     * constructor
     */
    public TransportDriver(TransportDriverConfig transportDriverConfig) {
        this.transportDriverConfig = transportDriverConfig;
    }

    /**
     * 启动
     */
    public synchronized void init() {
        if (started.compareAndSet(false, true)) {
            //
            singleThreadScheduler = new SingleThreadScheduler();
            singleThreadScheduler.init();
            //
            queueFactory = new KafkaTransportQueueFactory(transportDriverConfig.getQueueConfig());
            //
            transportDriverService = new DefaultTransportDriverService(queueFactory, singleThreadScheduler, transportDriverConfig);
            transportDriverService.init();
            //
        }
    }

    /**
     * 是否启动
     */
    public boolean isStarted() {
        return started.get();
    }

    /**
     * 销毁
     */
    public synchronized void destroy() {
        if (started.compareAndSet(true, false)) {
            //
            if (singleThreadScheduler != null){
                singleThreadScheduler.destroy();
            }
            //
            if (queueFactory != null){
                queueFactory.destroy();
            }
            //
            if (transportDriverService != null){
                transportDriverService.destroy();
            }
        }
    }
}
